package com.example.starter.work6

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently
import org.apache.rocketmq.common.consumer.ConsumeFromWhere
import org.apache.rocketmq.common.message.MessageExt
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel
import java.text.SimpleDateFormat
import java.time.LocalDateTime
import java.util.*


/**
@author whd
@date 2023/10/12 14:36:52
 */
fun main() {
  val consumer = DefaultMQPushConsumer("delay-test-0317", false)
  //设置nameserver
  consumer.namesrvAddr = "192.168.201.129:9876"
  //设置topic,subExpression即设置订阅的tag，*表示所有
  consumer.subscribe("delay-topic", "*")
  //从最新的offset拉取
  consumer.consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET
  consumer.messageModel = MessageModel.CLUSTERING
  //注册监听

  consumer.registerMessageListener(MessageListenerConcurrently { msgs: List<MessageExt?>?, context: ConsumeConcurrentlyContext? ->
    val currentTime = SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Date())
    System.out.printf("接收时间：%s %n", currentTime)
    ConsumeConcurrentlyStatus.CONSUME_SUCCESS
  })
  consumer.start()
  System.out.printf("Consumer Started.%n")
}



